package com.lupan.kafka;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * 说明：卡夫卡消费者线程类
 *
 * @author lupan
 * @version 2017/2/14 17:07
 */
public class MyKafkaConsumer implements Runnable{

    private static Logger logger = LoggerFactory.getLogger(MyKafkaConsumer.class);

    @Value("#{config('kafka.topic')}")
    private String topic;
    @Value("#{config('kafka.server')}")
    private String kafkaServer;

    private ConsumerConnector consumer;

    private Properties createKafkaConfig() {
        Properties props = new Properties();
        props.put("auto.offset.reset", "smallest");
        props.put("zookeeper.connect", kafkaServer);
        props.put("zookeeper.session.timeout.ms", "6000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("group.id", "ecu_version_sync_task");
        props.put("session.timeout.ms", "1000");
        props.put("auto.commit.enable", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("rebalance.max.retries", "4");
        props.put("rebalance.backoff.ms", "2000");
        return props;
    }

    @Override
    public void run() {
        try {
            consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(createKafkaConfig()));
            logger.info("kafkaConsumer线程启动成功！线程名：{}",Thread.currentThread().getName());
            Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
            topicCountMap.put(topic, 1);
            Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
            KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
            ConsumerIterator<byte[], byte[]> it = stream.iterator();
            while (!Thread.interrupted() && it.hasNext()) {
                logger.info("kafkaConsumer线程消费消息成功！消息内容：{}",new String(it.next().message(),"UTF-8"));
                try {
//                    ecuVerionService.updateEcuVersion(it.next().message());
                }catch (Exception e){
                    logger.error("处理kafka消息异常，异常信息：{}",e);
                }
            }
        } catch (Exception e){
            logger.error("kafkaConsumer线程启动失败！异常信息：", e);
        }finally {
            if (consumer != null) {
                logger.info("kafkaConsumer退出");
                consumer.shutdown();
            }
        }
    }


    public void stop() {
        if(consumer != null) {
            consumer.shutdown();
        }
    }
}
